package com.zhike;

import com.alibaba.fastjson.JSONObject;
import com.zhike.bean.Student;
import com.zhike.bean.TableProcess;
import com.zhike.bean.Teacher;
import com.zhike.bean.TempBean;
import com.zhike.constant.HbaseConstant;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.junit.Test;

import java.util.*;

public class test {
    public static void main(String[] args) {
        TableProcess tableProcess = new TableProcess("car", "dim_car", "id,name,age", "id,name", "");
        String sinkColumns = tableProcess.getSinkColumns();
        String[] columns = sinkColumns.split(",");
        String[] pks = tableProcess.getSinkPk().split(",");
        List<String> list = Arrays.asList(pks);
        StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                .append(HbaseConstant.HBASE_SCHEMA)
                .append(".")
                .append(tableProcess.getSinkTable())
                .append("(");
        for (int i = 0; i < columns.length; i++) {
            if (list.contains(columns[i])) {
                createTableSql.append(columns[i])
                        .append(" varchar not null");
            } else {
                createTableSql.append(columns[i])
                .append(" varchar");
            }
            if (i < columns.length - 1) {
                createTableSql.append(",");
            }
        }
        createTableSql.append(" constraint pk primary key (")
                .append(tableProcess.getSinkPk())
                .append(")")
                .append(tableProcess.getSinkExtend());
        System.out.println(createTableSql);

    }

    @Test
    public void test01(){
        HashMap<String, String> hashMap = new HashMap<>();
        hashMap.put("id", "1");
        hashMap.put("name", "tom");
        hashMap.put("age", "17");
        System.out.println(hashMap);
        Set<Map.Entry<String, String>> entries = hashMap.entrySet();
        System.out.println("entries = " + entries);
        Iterator<Map.Entry<String, String>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, String> next = iterator.next();
            if (next.getKey().equals("id")){
                iterator.remove();
            }
        }
        System.out.println(hashMap);
    }

    @Test
    public void test02(){
        String value="{\"before\":45,\"after\":{\"id\":14,\"name\":\"小王\",\"age\":\"15\"},\"ts\":24214123521}";
        JSONObject jsonObject = JSONObject.parseObject(value);
        System.out.println("jsonObject = " + jsonObject);
        JSONObject after = jsonObject.getJSONObject("after");
        System.out.println("after = " + after);
        Set<Map.Entry<String, Object>> entries = after.entrySet();
        System.out.println("entries = " + entries);
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            if (next.getKey().equals("id")){
                iterator.remove();
            }
        }
        System.out.println("entries = " + entries);
        System.out.println("after = " + after);
        System.out.println("jsonObject = " + jsonObject);

        Set<String> keySet = after.keySet();
        System.out.println("keySet = " + keySet);
        Collection<Object> values = after.values();
        System.out.println(values);
    }

    @Test
    public void broadcastDS(){
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //模范广播流中的数据
        DataStreamSource<Teacher> teacherDS = env.fromElements(new Teacher("1", "jack"));
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<String, String>("state",String.class,String.class);
        //模仿流中的数据
        DataStreamSource<Student> studentDS = env.fromElements(new Student("1", "tom", "1"));
        BroadcastStream<Student> broadcastDS = studentDS.broadcast(mapStateDescriptor);
        BroadcastConnectedStream<Teacher, Student> connect = teacherDS.connect(broadcastDS);
        SingleOutputStreamOperator<Tuple3<String, String, String>> process = connect.process(new BroadcastProcessFunction<Teacher, Student, Tuple3<String, String, String>>() {
            @Override
            public void processElement(Teacher value, BroadcastProcessFunction<Teacher, Student, Tuple3<String, String, String>>.ReadOnlyContext ctx, Collector<Tuple3<String, String, String>> out) throws Exception {
                ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                String a = broadcastState.get("a");
                System.out.println("a = " + a);
            }
            @Override
                public void processBroadcastElement(Student value, BroadcastProcessFunction<Teacher, Student, Tuple3<String, String, String>>.Context ctx, Collector<Tuple3<String, String, String>> out) throws Exception {
                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                broadcastState.put("a","b");
            }

//            //处理广播流
//            @Override
//            public void processBroadcastElement(Teacher value, BroadcastProcessFunction<Teacher, Student, Tuple3<String, String, String>>.Context ctx, Collector<Tuple3<String, String, String>> out) throws Exception {
//                //获取广播状态并广播出去
//                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
//                System.out.println("broadcastState = " + broadcastState);
//                String id = value.getId();
//                System.out.println("id = " + id);
//                String name = value.getName();
//                System.out.println("name = " + name);
//                broadcastState.put(id, name);
//                System.out.println(broadcastState.get(id));
//                out.collect(new Tuple3<String, String, String>(id, name, name));
//            }
//
//            //处理主流
//            @Override
//            public void processElement(Student value, BroadcastProcessFunction<Student, Teacher, Tuple3<String, String, String>>.ReadOnlyContext ctx, Collector<Tuple3<String, String, String>> out) throws Exception {
//                //获取广播状态并根据广播状态处理主流数据
//                ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
//                System.out.println("broadcastState = " + broadcastState);
//                System.out.println(broadcastState.get("1"));
//                String teacherId = value.getTeacher_id();
//                System.out.println("teacherId = " + teacherId);
//                String teacherName = broadcastState.get(teacherId);
//                System.out.println(teacherName);
//                out.collect(new Tuple3<String, String, String>(value.getId(), value.getName(), teacherName));
//            }

        });

        process.print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }


    @Test
    public void testBuilder(){
        System.out.println(25/10);
    }
}

